/**
 * @file StreamPlayer.c
 *
 * This file implements the StreamPlayer interface.
 *
 * @author Luc Fourestier
 * @date 2009
 *
 * @page SP The StreamPlayer
 *
 * The StreamPlayer reads the tagged element out of the circular buffer and send or store them accordingly.
 *
 * @section SP_Design The design
 *
 * The StreamPlayer is created via SP_Create() using SP_Options_t as input.
 *
 * TO BE COMPLETED
 *
 */

#include "StreamPlayer.h"

#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>

#define GUARD_TIME_US 50 //!< Used in the jitter algorithm to save CPU
#define SOCKET_SNDBUFF_SIZE 150000

// Statics
static void * PlayerThread(void * Arg);
static int DumpToFile(SP_StreamPlayer_t * SP, unsigned char *Buffer, unsigned int Size);
static int StreamToUDP(SP_StreamPlayer_t * SP, unsigned char *Buffer, unsigned int Size, struct timeval TimeToSend);

/**
 * Create a StreamPlayer.
 *
 * @param SP_p (out): The created StreamPlayer.
 * @param Options (in): The options of creation.
 * @return Errors
 */
int SP_Create(SP_StreamPlayer_t ** SP_p, SP_Option_t Options)
{
	int RetVal = OK;
	SP_StreamPlayer_t * SP;

	if (SP_p != NULL)
	{
		unsigned int opt = 1;
		unsigned int OptSize = sizeof(opt);

		// Create the SP
		SP = (SP_StreamPlayer_t*) malloc(sizeof(SP_StreamPlayer_t));
		if (SP == NULL)
		{
			return(SP_ERROR_CANNOT_ALLOCATE);
		}
		bzero((void*)SP, sizeof(SP_StreamPlayer_t));

		// remember options
		SP->Options = Options;

		// Ouput type
		switch (SP->Options.OutputType)
		{
			case File:
				SP->StorageFile = fopen(SP->Options.FileName, "w+");
				if (SP->StorageFile == NULL)
				{
					return(SP_ERROR_CANNOT_OPEN_FILE);
				}
				break;

			case Udp:
				SP->Socket = socket(AF_INET, SOCK_DGRAM, 0);
				if (SP->Socket < 0)
				{
					return(SP_ERROR_SOCKET_ERROR);
				}

				if (setsockopt(SP->Socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1)
				{
					return(SP_ERROR_SOCKET_ERROR);
				}

				opt = SOCKET_SNDBUFF_SIZE;
				if (setsockopt(SP->Socket, SOL_SOCKET, SO_RCVBUF, (void*) &opt, OptSize) == -1)
				{
					return(SP_ERROR_SOCKET_ERROR);
				}

				bzero((char *) &SP->RemoteUDPAddr, sizeof(SP->RemoteUDPAddr));
				SP->RemoteUDPAddr.sin_family = AF_INET;
				SP->RemoteUDPAddr.sin_addr.s_addr = inet_addr(SP->Options.SendAddress);
				SP->RemoteUDPAddr.sin_port = htons(SP->Options.SendPort);
				break;

			default:
				return(SP_ERROR_NOT_SUPPORTED);
				break;
		}

		// Init the analyzer thread
		SP->ThreadLoop = 1;
		if (pthread_create(&SP->PlayerThreadID, NULL, PlayerThread, (void*) SP) != 0)
		{
			return(SP_ERROR_CANNOT_ALLOCATE);
		}

		// Return the structure
		*SP_p = SP;
	}
	else
	{
		RetVal = SP_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * Delete the StreamPlayer
 *
 * @param SP (in): The StreamPlayer to delete.
 * @return Errors
 */
int SP_Delete(SP_StreamPlayer_t * SP)
{
	int RetVal = OK;

	if (SP != NULL)
	{
		// Kill threads
		SP->ThreadLoop = 0;
		CircB_KillRead(SP->Options.CB);
		pthread_join(SP->PlayerThreadID, NULL);

		switch (SP->Options.OutputType)
		{
			case File:
				// Close Storage file
				if (SP->StorageFile != NULL)
				{
					fclose(SP->StorageFile);
					SP->StorageFile = NULL;
				}
				break;

			case Udp:
				// Close UDP transmission
				if (SP->Socket != 0)
				{
					shutdown(SP->Socket, SHUT_RDWR);
					close(SP->Socket);
					SP->Socket = 0;
				}
				break;

			default:
				break;
		}

		// Delete
		free(SP);
	}
	else
	{
		RetVal = SP_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * Print the StreamPlayer info structure
 *
 * @param SP (in): The SP to print
 * @param Name (in): The name of the SP (any nickname you wish!).
 * @param level (in): Level of details (Lowest = 0).
 * @param Output (in): Where to print (stdout or FILE).
 * @return Errors
 */
int SP_Print(SP_StreamPlayer_t * SP, char * Name, int Level, FILE * Output)
{
	int RetVal = OK;

	if ((SP == NULL) && (Output == NULL))
	{
		return(ERROR);
	}

	if (Name == NULL)
	{
		Name = "StreamPlayer";
	}

	switch (SP->Options.OutputType)
	{
	case File:
		fprintf(Output, "%s ouput file: %s\n", Name, SP->Options.FileName);
		break;

	case Udp:
		fprintf(Output, "%s output address: %s:%i\n", Name, SP->Options.SendAddress, SP->Options.SendPort);
		break;

	default:
		break;
	}

	fprintf(Output, "%s total bytes played: %lli\n", Name, SP->TotalBytePlayed);

	// Internals status
	if (Level >= PRINT_LEVEL_INTERNALS)
	{
		fprintf(Output, "%s threads state: %i\n", Name, (int)SP->ThreadLoop);
		fprintf(Output, "%s total dump error: %lli\n", Name, SP->TotalDumpError);
		fprintf(Output, "%s total UDP error: %lli\n", Name, SP->TotalUDPError);
#ifdef DEBUG_SP_PROFILING
		if (SP->passCount > 1)
		{
			fprintf(Output, "%s total pass duration: %is %iµs\n", Name, (int)SP->TotalPassDuration.tv_sec, (int)SP->TotalPassDuration.tv_usec);
			fprintf(Output, "%s pass count: %lli\n", Name, SP->passCount-1);
			fprintf(Output, "%s Average duration per pass: %lli µs\n", Name, TIMEVAL_TO_USECONDS(&SP->TotalPassDuration)/(SP->passCount-1));
			fprintf(Output, "%s pass errors: %lli\n", Name, SP->passError);
		}
#endif
	}

	fprintf(Output, "\n");

	return(RetVal);
}

/**
 * Dump the stream.
 *
 * @param SP (in): The StreamPlayer
 * @param Buffer (in): The new chunk buffer to treat
 * @param Size (in): The size of the buffer
 * @return Errors
 */
static int DumpToFile(SP_StreamPlayer_t * SP, unsigned char *Buffer, unsigned int Size)
{
	int RetVal = OK;

	if ((SP != NULL) && (Buffer != NULL) && (Size > 0))
	{
		// Store the stream
		if (SP->StorageFile != NULL)
		{
			if (fwrite((void *) Buffer, Size, 1, SP->StorageFile) == 0)
			{
				RetVal = SP_ERROR_CANNOT_DUMP;
				SP->TotalDumpError++;
			}
			else
			{
				SP->TotalBytePlayed += Size;
			}
		}
	}
	else
	{
		RetVal = SP_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * Transmit the stream to UDP.
 *
 * We introduced usleep() to save CPU, but we do not want it to deteriorate our time precision.
 * That's why we introduced the guard time GUARD_TIME_US (to wake up a bit before).
 *
 * @param SP (in): The StreamPlayer
 * @param Buffer (in): The new chunk buffer to treat
 * @param Size (in): The size of the chunk buffer
 * @param TimeToSend (in): The time stamp when to send that chunk
 * @return Errors
 */
static int StreamToUDP(SP_StreamPlayer_t * SP, unsigned char *Buffer, unsigned int Size, struct timeval TimeToSend)
{
	int RetVal = OK;
	struct timeval NewTime;

	if ((SP != NULL) && (Buffer != NULL) && (Size > 0) && (SP->Socket > 0))
	{

		// Wait before sending the next packet
		do
		{
			struct timeval TempTime;
			TempTime.tv_sec = 0;
			TempTime.tv_usec = GUARD_TIME_US;

			gettimeofday(&NewTime, NULL);
			timersub(&NewTime, &SP->Options.ReferenceTime, &NewTime); // Relative time

			// Sleep as long as it can but within a guard margin (to avoid CPU usage)
			timersub(&TimeToSend, &TempTime, &TempTime);
			if (timercmp(&NewTime, &TempTime, <))
			{
				timersub(&TempTime, &NewTime, &TempTime);
				usleep(TempTime.tv_usec+(TempTime.tv_sec*1000000));
			}
			else
			{
				usleep(1);
			}
		} while (timercmp(&NewTime, &TimeToSend, <));

		// Transmit the stream over UDP
		int Count = sendto(SP->Socket, (void *)Buffer, Size, 0, (struct sockaddr*) &SP->RemoteUDPAddr, (socklen_t) sizeof(struct sockaddr_in));
		if (Count <= 0)
		{
			RetVal = SP_ERROR_SOCKET_ERROR;
			SP->TotalUDPError++;
		}
		else
		{
			SP->TotalBytePlayed += Count;
		}
	}
	else
	{
		RetVal = SP_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * The player thread.
 * The player which take the actual action on the tagged chunks (Send, dump, etc...).
 *
 * @param Arg: the SP_StreamPlayer_t structure
 */
static void * PlayerThread(void * Arg)
{
	SP_StreamPlayer_t * SP = (SP_StreamPlayer_t *) Arg;
	Tag_t * Tag = NULL;
	void * Buff_p = NULL;
	int Error = OK;

	if (SP == NULL)
	{
		pthread_exit(NULL);
		return(NULL);
	}


#ifdef DEBUG_SP_PROFILING
		struct timeval Start;
		Start.tv_sec = 0;
		Start.tv_usec = 0;
#endif

	while (SP->ThreadLoop)
	{
		unsigned int Size = 0;

#ifdef DEBUG_SP_PROFILING
			struct timeval New;
			gettimeofday(&New, NULL);
			if (Start.tv_sec == 0)
			{
				Start = New;
			}
			timersub(&New, &Start, &SP->TotalPassDuration);
			SP->passCount++;
#endif

		// Treat the stream
		Error = CircB_GetReadElt(SP->Options.CB, &Buff_p, &Size, CIRCB_WAIT_FOREVER);
		if (Error == OK)
		{
			Tag = (Tag_t*)Buff_p;
			Size -= sizeof(Tag_t);

			if (Tag->ReadyToPlay == 1)
			{
				switch (SP->Options.OutputType)
				{
				case File:
					Error = DumpToFile(SP, ((unsigned char*)Buff_p)+sizeof(Tag_t), Size);
					break;
				case Udp:
					Error = StreamToUDP(SP, ((unsigned char*)Buff_p)+sizeof(Tag_t), Size, Tag->TimeToSend);
					break;

				default:
					break;
				}

				Error = CircB_ValidateReadElt(SP->Options.CB);
			}
			else
			{
				Error = CircB_ReleaseReadElt(SP->Options.CB);
			}
		}

#ifdef DEBUG_SP_PROFILING
		if (Error != OK)
		{
			SP->passError++;
		}
		Error = OK;
#endif
	}

	pthread_exit(NULL);
}

